// Copyright 2023 Srhino Co., Ltd. All rights reserved.

package v1

import (
	"context"
	"errors"
	"git.ouryun.cn/lzj1/grpc-bidirectional-stream/pkg/apis/v1"
	v12 "git.ouryun.cn/lzj1/grpc-bidirectional-stream/pkg/interfaces/v1"
	"git.ouryun.cn/lzj1/grpc-bidirectional-stream/pkg/utils"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/proto"
	"log"
	"sync/atomic"
	"time"
)

type grpcClient struct {
	address    string
	streamData *StreamData
	stream     v1.StreamDataService_UserStreamClient
	callback   RequestProcessFunc
	ctx        context.Context
}

var _ v12.GrpcServer = (*grpcClient)(nil)

func NewGrpcClient(addr string, ctx context.Context, callback RequestProcessFunc) v12.GrpcServer {
	return &grpcClient{
		address:    addr,
		streamData: NewStreamData(),
		callback:   callback,
		ctx:        ctx,
	}
}

func (s *grpcClient) Request(data proto.Message, params *v1.QueryParams, messageType v1.StreamMessage_Type, requestId uint64) (*v1.StreamMessage, error) {
	dataAny, err := utils.MessageToAny(data)
	if err != nil {
		return nil, err
	}

	message := &v1.StreamMessage{
		Type:   messageType,
		Data:   dataAny,
		Params: params,
	}

	if messageType == v1.StreamMessage_REQUEST_TYPE {
		atomic.AddInt64(&reqId, 1)
		message.RequestId = uint64(reqId)
	} else {
		message.RequestId = requestId
	}

	s.streamData.Requests <- message
	if messageType == v1.StreamMessage_REQUEST_TYPE {
		for {
			select {
			case <-time.After(5 * time.Second):
				return nil, errors.New("request timeout")
			case msg := <-s.streamData.Responses:
				if msg.RequestId == message.RequestId {
					return msg, nil
				}
			}
		}
	}
	return nil, nil
}

// Start
// TODO 后续优化加上重连机制
func (s *grpcClient) Start() error {

	err := s.connection()
	if err != nil {
		return err
	}

	// 处理转发请求
	go func() {
		s.forwardRequest()
	}()

	// 接收服务端返回的数据
	waitc := make(chan struct{})
	if err := s.handleRecv(); err != nil {
		log.Println(err)
		return err
	}
	<-waitc
	return nil
}

func (s *grpcClient) connection() error {
	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	}
	conn, err := grpc.Dial(s.address, opts...)
	if err != nil {
		log.Println(err)
		return err
	}
	//defer conn.Close()
	log.Println("grpc client start")
	client := v1.NewStreamDataServiceClient(conn)

	stream, err := client.UserStream(s.ctx)
	if err != nil {
		log.Println(err)
		return err
	}

	s.stream = stream
	return nil
}

// 接收gRPC服务端数据
func (s *grpcClient) handleRecv() error {
	for {
		recv, err := s.stream.Recv()
		log.Printf("接收到服务端数据 %v", recv)
		if err != nil {
			// TODO 这里只是简单做了下断线重连，后续可以加上重连机制
			if grpc.Code(err) == codes.Unavailable {
				time.Sleep(5 * time.Second)
				if err := s.connection(); err != nil {
					return err
				}
				continue
			}
			return err
		}
		// 服务端数据处理
		if recv.Type == v1.StreamMessage_RESPONSE_TYPE {

			if err := s.handleResponse(recv); err != nil {
				return err
			}
		} else {
			s.handleRequest(recv)
		}
	}
}

// 处理grpc客户端响应数据
func (s *grpcClient) handleResponse(recv *v1.StreamMessage) error {
	log.Printf("handleResponse %v", recv)
	s.streamData.Responses <- recv
	return nil
}

// 处理grpc客户端请求数据
func (s *grpcClient) handleRequest(recv *v1.StreamMessage) {
	log.Printf("handleRequest %v", recv)
	if s.callback != nil {
		s.callback(recv)
	}
}

// 处理转发请求
func (s *grpcClient) forwardRequest() {
	for {
		select {
		case request := <-s.streamData.Requests:
			log.Println(request)
			if err := s.sendMessage(request); err != nil {
				log.Println(err)
			}
		}
	}

}

// 发送消息到grpc服务端
func (s *grpcClient) sendMessage(message *v1.StreamMessage) error {
	if s.stream == nil {
		return errors.New("stream closed")
	}
	log.Printf("发送消息 %v", message)
	err := s.stream.Send(message)

	if err != nil {
		return err
	}

	return nil
}

func (s *grpcClient) Stop() {
	// TODO implement me
}
